/**
 * Copyright (c) 2023 murenchao
 * taomu is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *       http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 */
package cool.taomu.utils

import java.net.InetSocketAddress
import java.util.concurrent.ExecutorService
import org.apache.thrift.TProcessor
import org.apache.thrift.TProcessorFactory
import org.apache.thrift.protocol.TCompactProtocol
import org.apache.thrift.server.THsHaServer
import org.apache.thrift.server.TThreadedSelectorServer
import org.apache.thrift.transport.TNonblockingServerSocket
import org.apache.thrift.transport.TSocket
import org.apache.thrift.transport.layered.TFramedTransport
import org.slf4j.LoggerFactory

class ThriftUtils {
	static val LOG = LoggerFactory.getLogger(ThriftUtils);

	def static startServer(TProcessor tprocessor, InetSocketAddress isa, ExecutorService pool, int selector) {
		var TNonblockingServerSocket serverTransport = null;
		try {
			serverTransport = new TNonblockingServerSocket(isa);
			var args = new TThreadedSelectorServer.Args(serverTransport);
			//args.processor(tprocessor);
			args.protocolFactory(new TCompactProtocol.Factory());
			args.transportFactory(new TFramedTransport.Factory());
			args.processorFactory(new TProcessorFactory(tprocessor));
			//args.selectorThreads(selector === 1 ? 2 : selector);
			args.selectorThreads(4);
			args.workerThreads(Runtime.getRuntime().availableProcessors() * 2);
			//args.executorService(pool);
			args.acceptPolicy(TThreadedSelectorServer.Args.AcceptPolicy.FAST_ACCEPT);
			args.acceptQueueSizePerThread(8);
			var server = new TThreadedSelectorServer(args);
			System.out.println("start server....");
			server.serve();
		} catch (Exception ex) {
			if (serverTransport !== null) {
				serverTransport.close();
			}
		}
	}

	def static startServer(TProcessor tprocessor, InetSocketAddress isa, int max, int min) {
		var TNonblockingServerSocket socket = null;
		try {
			socket = new TNonblockingServerSocket(isa);
			var arg = new THsHaServer.Args(socket).minWorkerThreads(min).maxWorkerThreads(max);
			arg.protocolFactory(new TCompactProtocol.Factory());
			arg.transportFactory(new TFramedTransport.Factory());
			arg.processorFactory(new TProcessorFactory(tprocessor));
			var server = new THsHaServer(arg);
			System.out.println("start server....");
			server.serve();
		} catch (Exception ex) {
			if (socket !== null) {
				socket.close();
			}
		}
	}

	static interface TRpc {
		def void callback(TCompactProtocol protocol);
	}

	def synchronized static void rpc(String host, int port, TRpc rpc) {
		LOG.info("host:{}", host);
		try {
			try(var tTransport = new TFramedTransport(new TSocket(host, port))) {
				var protocol = new TCompactProtocol(tTransport);
				if (tTransport !== null && !tTransport.isOpen()) {
					tTransport.open();
				}
				if (rpc !== null) {
					rpc.callback(protocol);
				}
			}
		} catch (Exception ex) {
			LOG.info("client exception:", ex)
		}
	}
}
